import { Kafka, logLevel, Partitioners } from 'kafkajs';
import { v4 as uuidv4 } from 'uuid';
import logger from '../Logger/logger';
import { createOrderService } from '../service/OrderService';

const kafka = new Kafka({
  clientId: uuidv4(),
  brokers: [process.env.KAFKA_BROKER || '127.0.0.1:9092'], // 使用环境变量配置 Kafka broker 地址
  logLevel:logLevel.ERROR,
});

const admin = kafka.admin();

const producer = kafka.producer({
  createPartitioner: Partitioners.LegacyPartitioner,
});

async function isQueueFull(topic: string, threshold: number): Promise<boolean> {
  await admin.connect();
  const topicOffsets = await admin.fetchTopicOffsets(topic);
  const consumerOffsets = await admin.fetchOffsets({ groupId: 'test-group', topics: [topic] });

  let isFull = false;
  topicOffsets.forEach((partitionOffset, index) => {
    const consumerPartition = consumerOffsets[0].partitions.find(p => p.partition === partitionOffset.partition);
    if (consumerPartition) {
      const lag = parseInt(partitionOffset.high) - parseInt(consumerPartition.offset);
      if (lag > threshold) {
        isFull = true;
      }
    }
  });

  await admin.disconnect();
  return isFull;
}

async function processOrQueueMessage(topic: string, message: string, threshold: number, fn) {
  const queueFull = await isQueueFull(topic, threshold);
  if (queueFull) {
    logger.warn('Queue is full, processing message directly');
    await fn && fn(message);
  } else {
    await producer.send({
      topic,
      messages: [{ value: message }],
    });
  }
}


const consumer = kafka.consumer({ groupId: 'test-group' });

producer.on('producer.connect', () => {
  console.log('Kafka Producer connected');
});

producer.on('producer.disconnect', () => {
  console.log('Kafka Producer disconnected');
});

producer.on('producer.network.request_timeout', (err) => {
  console.error('Kafka Producer request timeout:', err);
});

producer.on('producer.network.request', () => {
  console.error('Kafka Broker Connection Refused:');
});

async function connectProducer() {
  try {
    await producer.connect();
    console.log('Kafka Producer connected');
  } catch (err) {
    console.error('Kafka Producer connection error:', err);
  }
}


async function runConsumer(topic: string, fn?: Function, groupId: string = 'test-group') {
  try {
    await consumer.subscribe({ topic, fromBeginning: true });
    logger.info('Consumer is ready to consume messages：' + topic);
    await consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        try {
          logger.info('Received message：' + message.value.toString());
          fn && fn(message.value.toString());
        } catch (error) {
          logger.error('Error consuming message:', error);
        }
      },
    });
  } catch (error) {
    logger.error('消费失败', error);
  }
}

runConsumer('orderTopic', async (val: any) => {
  console.log(val);
  await createOrderService(val);
}).catch(err => {
  logger.error('消费失败', err);
});


export { producer, consumer, connectProducer, runConsumer, processOrQueueMessage };